/*
 * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
 * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
 */
package com.excesys.exsecs.config;

import com.excesys.exsecs.handler.MqttCallbackHandle;
import com.excesys.exsecs.linkage.handle.RulesCallbackHandle;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.handler.annotation.Header;

/**
 * @author XuYangbo
 */
@Configuration
public class MqttConfig {

	@Autowired
	private MqttEntity mqttEntity;

	@Autowired
	private MqttCallbackHandle mqttCallbackHandle;

	@Autowired
	private RulesCallbackHandle rulesCallbackHandle;

	public static final String CHANNEL_NAME_IN = "mqttInboundChannel";

	public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";

	private static final byte[] WILL_DATA;

	static {
		WILL_DATA = "offline".getBytes();
	}

	public MqttConnectOptions getMqttConnectOptions() {
		MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
		mqttConnectOptions.setCleanSession(true);
		mqttConnectOptions.setUserName(mqttEntity.getUsername());
		mqttConnectOptions.setPassword(mqttEntity.getPassword().toCharArray());
		mqttConnectOptions.setServerURIs(mqttEntity.getUrl().split(","));
		mqttConnectOptions.setConnectionTimeout(30);
		mqttConnectOptions.setKeepAliveInterval(60);
		mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
		// 支持同时发送消息数为1000，默认为10
		mqttConnectOptions.setMaxInflight(1000);
		return mqttConnectOptions;
	}

	public MqttConnectOptions getMqttConnectOptions2() {
		MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
		mqttConnectOptions.setCleanSession(true);
		mqttConnectOptions.setUserName(mqttEntity.getUsername());
		mqttConnectOptions.setPassword(mqttEntity.getPassword().toCharArray());
		mqttConnectOptions.setServerURIs(mqttEntity.getUrl().split(","));
		mqttConnectOptions.setConnectionTimeout(30);
		mqttConnectOptions.setKeepAliveInterval(60);
		mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
		mqttConnectOptions.setMaxInflight(1000);
		return mqttConnectOptions;
	}

	@Bean
	public MqttPahoClientFactory mqttClientFactory() {
		DefaultMqttPahoClientFactory defaultMqttPahoClientFactory = new DefaultMqttPahoClientFactory();

		MqttConnectOptions mco = getMqttConnectOptions();

		defaultMqttPahoClientFactory.setConnectionOptions(mco);
		return defaultMqttPahoClientFactory;
	}

	@Bean
	public MqttPahoClientFactory mqttClientFactory2() {
		DefaultMqttPahoClientFactory defaultMqttPahoClientFactory = new DefaultMqttPahoClientFactory();

		MqttConnectOptions mco = getMqttConnectOptions2();

		defaultMqttPahoClientFactory.setConnectionOptions(mco);
		return defaultMqttPahoClientFactory;
	}

	@Bean(name = CHANNEL_NAME_OUT)
	public MessageChannel mqttOutBoundChannel() {
		return new DirectChannel();
	}

	@Bean
	@ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
	public MessageHandler mqttOutBound() {
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(MqttClient.generateClientId(),
				mqttClientFactory());
		messageHandler.setAsync(true);
		messageHandler.setDefaultTopic(mqttEntity.getProducerDefaultTopic());
		return messageHandler;
	}

	@MessagingGateway(defaultRequestChannel = CHANNEL_NAME_OUT)
	public interface MqttProducer {

		void sendToMqtt(String payload);

		void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

		void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);

	}

	@Bean
	public MessageProducer inBound() {
		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
				MqttClient.generateClientId(), mqttClientFactory2(),
				mqttEntity.getConsumerDefaultTopic().split(","));

		// adapter.addTopic(DefContants.CONSUMER_TEST_TOPIC);
		// adapter.addTopic(DefContants.CONSUMER_SYS_CLIENTS_TOPIC);
		adapter.setCompletionTimeout(5000);
		adapter.setConverter(new DefaultPahoMessageConverter());
		adapter.setQos(1);

		adapter.setOutputChannel(mqttInBoundChannel());
		return adapter;
	}

	@Bean(name = CHANNEL_NAME_IN)
	public MessageChannel mqttInBoundChannel() {
		return new DirectChannel();
	}

	@Bean
	@ServiceActivator(inputChannel = CHANNEL_NAME_IN)
	public MessageHandler handler() {
		return message -> {
			String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
			String payload = message.getPayload().toString();
			mqttCallbackHandle.handle(topic, payload);
			rulesCallbackHandle.handle(topic, payload);
		};
	}

}
